package com.nightsoul.hadoop1.test.dcache;

import java.io.File;
import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nightsoul.hadoop1.test.NcdcStationMetadata;
import com.nightsoul.hadoop1.test.OldJobBuilder;
import com.nightsoul.hadoop1.test.junit.NcdcRecordParser;
import com.nightsoul.hadoop1.test.mapreduce.MaxTemperatureReduce;

/**
 * 输出结果与预期不一致
 * @author zj
 *
 */
public class MaxTemperatureByStationNameUsingDistributedCacheFile extends Configured implements Tool {

	static class StationTemeratureMapper extends MapReduceBase
		implements Mapper<LongWritable, Text, Text, IntWritable> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, IntWritable> output, Reporter reporter)
				throws IOException {

			parser.parse(value);
			if(parser.isMissingTemperature()) {
				output.collect(new Text(parser.getStationId()), new IntWritable(parser.getAirTemperature()));
				
			}
		}
		
	}
	
	static class StationTemeratureReducer extends MapReduceBase
		implements Reducer<Text, IntWritable, Text, IntWritable> {
		private NcdcStationMetadata metadata = new NcdcStationMetadata();
		
		@Override
		public void configure(JobConf job) {
			super.configure(job);
			try {
				metadata.initialize(new File("stations-fixed-width.txt"));;
			} catch(IOException e) {
				throw new RuntimeException(e);
			}
		}
		
		@Override
		public void reduce(Text key, Iterator<IntWritable> values,
				OutputCollector<Text, IntWritable> output, Reporter reporter)
				throws IOException {
			String stationName = metadata.getStationName(key.toString());
			
			int maxValue = Integer.MIN_VALUE;
			while(values.hasNext()) {
				maxValue = Math.max(maxValue, values.next().get());
			}
			output.collect(new Text(stationName), new IntWritable(maxValue));
		}
		
	}
	
	@Override
	public int run(String[] args) throws Exception {
		JobConf conf = OldJobBuilder.parseInputAndOutput(this, getConf(), args);
		if(conf==null) {
			return -1;
		}
		
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(IntWritable.class);
		
		conf.setMapperClass(StationTemeratureMapper.class);
		conf.setCombinerClass(MaxTemperatureReduce.class);
		conf.setReducerClass(StationTemeratureReducer.class);
		
		JobClient.runJob(conf);
		return 0;
	}
	
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new MaxTemperatureByStationNameUsingDistributedCacheFile(), args);
		System.exit(exitCode);
	}

}
